package com.alison.datastream.chapter6_timeAndWaterMark.watermark;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author alison
 * @Date 2024/4/5 0:47
 * @Version 1.0
 * @Description
 */
public class W2_LatenData2 {
    /*

    输出
source: h1,1553503181000
source: h1,1553503182000
source: h1,1553503183000
source: h1,1553503184000
source: h1,1553503185000
source: h1,1553503186000
source: h1,1553503187000
source: h1,1553503188000
source: h1,1553503189000
source: h1,1553503190000
start: 1553503181000 1553503182000 1553503183000 1553503184000
(h1,4)
source: h1,1553503187000
source: h1,1553503184000
source: h1,1553503191000
source: h1,1553503192000
source: h1,1553503193000
source: h1,1553503194000
source: h1,1553503195000
start: 1553503185000 1553503186000 1553503187000 1553503188000 1553503189000 1553503187000
(h1,6)
source: h1,1553503196000
source: h1,1553503197000
source: h1,1553503198000
source: h1,1553503199000
source: h1,1553503200000
start: 1553503190000 1553503191000 1553503192000 1553503193000 1553503194000
(h1,5)
source: h1,1553503201000
source: h1,1553503202000
source: h1,1553503203000
source: h1,1553503204000
source: h1,1553503205000
start: 1553503195000 1553503196000 1553503197000 1553503198000 1553503199000
(h1,5)

     */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setMaxParallelism(1).setParallelism(1);
//        env.getConfig().setAutoWatermarkInterval(5000L);//tigger fun of onPeriodicEmit
//        SingleOutputStreamOperator<String> dataStream = env.socketTextStream("192.168.56.101", 12345);
        DataStreamSource<String> dataStream = env.addSource(new MySourceForLaten());
        DataStream source = dataStream.map((ele) -> {
            String[] split = ele.split(",");
            return Tuple2.of(split[0], Long.parseLong(split[1].trim()));
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));
        // 直到 数据生成的watermark到达 窗口的endTs的时候才触发,
        // 看结果，1553503181000-1553503190000，才触发，提取出1553503181000-1553503184000的元素进行聚合
        SingleOutputStreamOperator<Tuple2<String, Long>> input = source.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5)) //允许乱序数据
                        .withTimestampAssigner(((element, recordTimestamp) -> element.f1))); // 会丢失 1553503184000, 不会丢失 1553503187000

        // allowedLateness 允许延迟5s, 每个元素的eventTime跟watermark的时间比较，在5s的延迟内的元素，来一次，触发一下windows类的计算，超过这些数据，丢弃
        // 没有配置延迟水印，发送延迟的数据直接丢弃
        input.keyBy(item -> item.f0).window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                .allowedLateness(Duration.ofSeconds(5))//允许延迟5s ,不会丢失 1553503184000, 接触到1553503184000 就触发
                .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>() {
                    public void process(String key, Context context, Iterable<Tuple2<String, Long>> elements, Collector<Tuple2<String, Long>> out) throws Exception {
                        long count = 0;
                        StringBuilder tsList = new StringBuilder();
                        for (Tuple2<String, Long> element : elements) {
                            count += 1;
                            tsList.append(element.f1 + " ");
                        }
                        System.out.println("start: " + tsList);
                        out.collect(Tuple2.of(key, count));
                    }
                }).print();
        env.execute();
    }
}
